package a_aa_amainpackage.a_o2odata_deal.utils

import java.text.SimpleDateFormat

import a_aa_amainpackage.a_o2odata_deal.config.FieldsConfig._
import a_aa_amainpackage.a_o2odata_deal.config.config._
import com.alibaba.fastjson.{JSON, JSONObject}
import org.apache.spark.sql.{DataFrame, SQLContext}
/*
*项目名: o2odata_commend
*文件名: price_deal
*创建者: hancongcong
*创建时间:2019/8/14 11:51
*描述: TODO
*/ object price_deal {

  def pricedeal(priceText: String): String = {
    var price = "-1"
    if (!priceText.equals("-1")) {
      if (priceText.contains("-")) {
        price = priceText.split("-")(0)
      } else {
        price = priceText
      }
    } else {
      price = "-1"
    }
    price
  }

  def price_restart(sqlContext: SQLContext, vardata: DataFrame):DataFrame = {
    //异常价格的处理
    val exec_data = exec_price(sqlContext,vardata)
    //关联上的
    val price1 = sp_price(sqlContext,exec_data)._1
    //关联不上的
    val price2 = zhekou_price(sqlContext,sp_price(sqlContext,exec_data)._2)

    /**
      * 价格结果
       */
    val value = price1.select("good_id", "sellCount", "priceText","salesAmount")
      .union(price2.select("good_id", "sellCount", "priceText","salesAmount"))
      .toDF("good_id", "sellCount", "priceText","salesAmount").where("cast(salesAmount as double)>0")
      .dropDuplicates("good_id")

    value
  }

  /**
    * 异常价格处理
    * @param sqlContext
    * @param vardata
    * @return
    */
  def exec_price(sqlContext: SQLContext, vardata: DataFrame): DataFrame = {

    sqlContext.udf.register("pricedeal", pricedeal _)
    //变量
    vardata.registerTempTable("vardata")
    //    println("原始数据条数：" + vardata.count())
    //取价格>0的数据
    val sellvardata = sqlContext.sql(
      """
        |select
        |*,
        |cast(pricedeal(priceText) as double)  price
        |from
        |vardata
      """.stripMargin).drop("priceText")
      .withColumnRenamed("price", "priceText")

    ////////////////////////////////处理extendedData
    val price_data = sellvardata.toJSON.rdd.map(line => {
      //取出价格
      val nObject: JSONObject = JSON.parseObject(line)
      var price = nObject.getOrDefault("priceText","-1").toString.toDouble

      //取出extendedData中的money,进行价格替换
      val string: String = nObject.getOrDefault("extendedData","-1").toString

      if (string!="-1"){
        val nObject1: JSONObject = JSON.parseObject(string)
        val string1: String = nObject1.getOrDefault("firePhoenixExtending","-1").toString
        if (string1!="-1"){
          val nObject2: JSONObject = JSON.parseObject(string1)
          val money = pricedeal(nObject2.getOrDefault("money","-1").toString).toDouble
          if (money!=(-1)){
            price = money
          }
        }
      }

      nObject.put("priceText",price)
      nObject.toString()
    })

    val frame: DataFrame = sqlContext.read.json(price_data).drop("extendedData")



    val value = frame.where("priceText > 0")
    println("价格大于0数据条数: " + value.count())
    value.registerTempTable("sellvardata")
    //    println("查看价格=0 的数据：")
    //    val values = sellvardata.where("priceText = 0")
    //    println(values.count())
    //
    //
    //    println("看是否存在满减==")
    //    sellvardata.where("coupon rlike '满' and coupon rlike '减'").show()
    //异常价格处理
    val execpricedata = sqlContext.sql(
      s"""
        |select
        |*,
        |${exec_prices}
        |from
        |sellvardata
      """.stripMargin).drop("priceText")
      .withColumnRenamed("priceTexts", "priceText")
    execpricedata.registerTempTable("execpricedata")
    val datas = sqlContext.sql(
      """
        |select
        |good_id,
        |sellCount,
        |price
        |from
        |(select
        |*,
        |cast(priceText as double) price
        |from
        |execpricedata)
        |order by price desc
        |limit 500
      """.stripMargin)
      .repartition(1).write.json(s"${exec_price_top}/${get_time()}")
    execpricedata
  }

  /**
    * 获取系统当前时间
    * @return
    */
  def get_time(): String = {
    val df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")
    val str = df.format(System.currentTimeMillis())
    str
  }

  /**
    * 补采价格处理
    * @param sqlContext
    */
  def  sp_price(sqlContext: SQLContext,exec_data:DataFrame): (DataFrame,DataFrame) = {
    sqlContext.udf.register("pricedeal",pricedeal _)
    //筛选完异常价格的数据
    exec_data.dropDuplicates("good_id")
    .registerTempTable("source_tmp")

    //采集的慢慢买的价格
    /*val manman_tmp1 = sqlContext.read.json(s"${pricesq_path}")
    val manman_tmp2 = sqlContext.read.json(s"s3a://o2o-sourcedata/obs-source-2019/8/tmall/tmall_sq_1908")
    val manman_tmp3 = sqlContext.read.json(s"s3a://o2o-sourcedata/obs-source-2019/7/tmall/tmall_sq_1907")*/

  //  val manman_tmp1 = sqlContext.read.json("s3a://o2o-sourcedata/obs-source-2019/9/tmall/tmall_sq_1909/*",
  //    "s3a://o2o-sourcedata/obs-source-2019/8/tmall/tmall_sq_1908/*",
  //    "s3a://o2o-sourcedata/obs-source-2019/7/tmall/tmall_sq_1907/*")

    //val manman_tmp = sqlContext.read.json(s"${pricesq_path}")
      //.dropDuplicates("good_id")

    val manman_tmp = sqlContext.read.json(s"${pricesq_path}",
      s"${lastmonth_pricesq_path}",
      s"${lastlastmonth_pricesq_path}")
    manman_tmp.registerTempTable("manman_tmp")
    val manman_tmp_clear1 = sqlContext.sql(
      """
        |select
        |*,
        |cast(pricedeal(explode_field.`priceText`) as double) as price
        |from
        |(select
        |*,
        |explode(add_to_field) explode_field
        |from
        |manman_tmp)
      """.stripMargin).where("price != '-1' and price != -1 and price > 0")
    manman_tmp_clear1.registerTempTable("manman_tmp_clear1")

    /////////测试代码无需保留，按平均值计算价格///////////////
    /*val manman_tmp_clear2 = sqlContext.sql(
      s"""
         |select
         |*,
         |row_number() over(partition by good_id order by avg_price asc) rank
         |from
         |(select
         |*,
         |case
         |when max_price-min_price>1000 then min_price
         |else sum/count end as avg_price
         |from
         |(select
         |*,
         |count(price) over(partition by good_id) as count,
         |sum(price) over(partition by good_id) as sum,
         |max(price) over(partition by good_id) as max_price,
         |min(price) over(partition by good_id) as min_price
         |from manman_tmp_clear1))
       """.stripMargin).where("rank  ='1'").drop("rank", "add_to_field", "platform_id",
      "explode_field","count","sum","price","max_price","min_price").withColumnRenamed("avg_price","price")
    manman_tmp_clear2.registerTempTable("manman_tmp_clear2")*/
    /////////测试代码无需保留，按平均值计算价格///////////////

    /////////原始代码需保留///////////////
    val manman_tmp_clear2 = sqlContext.sql(
      """
        |select
        |*,
        |row_number() over(partition by good_id order by price asc) rank
        |from
        |manman_tmp_clear1
      """.stripMargin).where("rank  ='1'").drop("rank", "add_to_field", "platform_id", "explode_field")
    manman_tmp_clear2.registerTempTable("manman_tmp_clear2")
    /////////原始代码需保留///////////////

    //关联慢慢买价格，关联上的用 原始价格和慢慢买比较 用较低价格
    val manman_tmp_clear3 = sqlContext.sql(
      """
        |select
        |t1.*,
        |case when cast(t2.price as double) > cast(t1.priceText as double) then t1.priceText else t2.price end as prices
        |from
        |source_tmp t1
        |join
        |manman_tmp_clear2 t2
        |on t1.good_id = t2.good_id
      """.stripMargin).dropDuplicates("good_id").drop("priceText", "salesAmountPre")
      .withColumnRenamed("prices", "priceText")
    manman_tmp_clear3.registerTempTable("manman_tmp_clear3")
    val manman_tmp_clear4 = sqlContext.sql(
      """
        |select
        |good_id,
        |sellCount,
        |shopId,
        |priceText,
        |(sellCount*priceText) as salesAmount
        |from
        |manman_tmp_clear3
      """.stripMargin)
//    manman_tmp_clear4.repartition(5).write.json(s"s3a://private/private_han/TM/2019/${currentmonth}/priceManmanmaiResData/1")
    //关联慢慢买价格，关联上的用慢慢买价格
    val manman_tmp_clear5 = sqlContext.sql(
      """
        |select
        |t1.good_id,
        |t1.sellCount,
        |t1.shopId,
        |t1.priceText
        |from
        |source_tmp t1
        |left join
        |manman_tmp_clear2 t2
        |on t1.good_id = t2.good_id
        |where t2.good_id is null
      """.stripMargin).dropDuplicates("good_id")
//    manman_tmp_clear5.repartition(5).write.json(s"s3a://private/private_han/TM/2019/${currentmonth}/priceManmanmaiResData/2")
    (manman_tmp_clear4,manman_tmp_clear5)
  }

  /**
    * 折扣价格处理
    * @param sqlContext
    * @return
    */
  def zhekou_price(sqlContext: SQLContext,sourcedf:DataFrame): DataFrame = {

    sourcedf.registerTempTable("sourcedf")
//    println("计算好销量价格的数据：" + sourcedf.count())
    val datas = sqlContext.sql(
      """
        |select
        |*,
        |cast(priceText as double) price
        |from
        |sourcedf
      """.stripMargin).drop("priceText").withColumnRenamed("price", "priceText")
    datas.registerTempTable("priceres")
    //  datas.show()
    //天猫优惠券计算
    println("优惠券计算：")
    //priceres
    //天猫店铺优惠券 shopId关联上的
    for (data <- Array(s"${shop_collection1}", s"${shop_collection2}", s"${shop_collection3}", s"${shop_collection4}")) {
      val tmsellzhekoou1 = sqlContext.read.json(s"s3a://o2o-sourcedata/obs-source-2020/${month}/${platform}/${data}/*")
        .where("promotion_info != '-1' and promotion_info is not null and promotion_info != '' and promotion_info != 'null' ")
      tmsellzhekoou1.registerTempTable("tmsellzhekoou1")
      //tmsellzhekoou1
      val tmshop1 = sqlContext.sql(
        """
          |select
          |t1.*,
          |t2.promotion_info
          |from
          |priceres t1
          |join
          |tmsellzhekoou1 t2
          |on t1.shopId = t2.shopId
        """.stripMargin)
      //折扣计算 公式 ((( 满/单价‘向上取整’) * 单价) - 减) / (( 满/单价‘向上取整’)*单价)
      //取出最低折扣
      val tmshop1zhekou = tmshop1.toJSON.rdd.map(line => {
        val nObject = JSON.parseObject(line)
        nObject.remove("_id")
        //解析 promotion_info
        val promotion_info_1 = nObject.getString("promotion_info")
        val priceText = nObject.getDouble("priceText")
        if (promotion_info_1.equals("-1")) {
          nObject.put("discount_re", "1")
        }
        else {
          val promotion_info = JSON.parseArray(promotion_info_1)
          val promotion_info_0 = JSON.parseObject(promotion_info.get(0).toString)
          val promotion_down = promotion_info_0.getDouble("promotion_down")
          val promotion_up = promotion_info_0.getDouble("promotion_up")
          val d = math.ceil(promotion_up / priceText)
          var discount_re = ((d * priceText - promotion_down) / (d * priceText)).formatted("%.4f")
          // 有promotion_pu 是 0 的情况，导致 discount_re 计算结果为 -Infinity
          if (promotion_up == 0) {
            discount_re = ((priceText - promotion_down) / priceText).formatted("%.4f")
          }
          for (i <- 1 to promotion_info.size() - 1) {
            val promotion_info_i = JSON.parseObject(promotion_info.get(i).toString)
            val promotion_down_i = promotion_info_i.getDouble("promotion_down")
            val promotion_up_i = promotion_info_i.getDouble("promotion_up")
            val di = math.ceil(promotion_up_i / priceText)
            var discount_re_i = ((di * priceText - promotion_down_i) / (di * priceText)).formatted("%.4f")
            // 有promotion_pu 是 0 的情况，导致 discount_re 计算结果为 -Infinity ((( 满/单价‘向上取整’) * 单价) - 减) / (( 满/单价‘向上取整’)*单价)
            if (promotion_up_i == 0) {
              discount_re_i = ((priceText - promotion_down_i) / priceText).formatted("%.4f")
            }
            if (discount_re_i < discount_re) {
              discount_re = discount_re_i
            }
          }
          //将优惠券的折扣计算出来
          nObject.put("discount_re", discount_re)
        }
        nObject.toJSONString
      })
      val frame = sqlContext.read.json(tmshop1zhekou).where("discount_re != 'NaN' and discount_re != '' and discount_re is not null ").select("discount_re", "good_id")
      frame.registerTempTable(s"zhekou_${data}")
    }
    //    priceres "20181108","20181116","20181124","20181130"
    val zhekoudata = sqlContext.sql(
      s"""
         |select
         |t1.*,
         |case when t2.good_id is null then '1' else t2.discount_re end discount_re_1,
         |case when t3.good_id is null then '1' else t3.discount_re end discount_re_2,
         |case when t4.good_id is null then '1' else t4.discount_re end discount_re_3,
         |case when t5.good_id is null then '1' else t5.discount_re end discount_re_4
         |
         |from
         |priceres t1
         |left join
         |zhekou_${shop_collection1} t2
         |on t1.good_id = t2.good_id
         |left join
         |zhekou_${shop_collection2} t3
         |on t1.good_id = t3.good_id
         |left join
         |zhekou_${shop_collection3} t4
         |on t1.good_id = t4.good_id
         |left join
         |zhekou_${shop_collection4} t5
         |on t1.good_id = t5.good_id
      """.stripMargin).dropDuplicates("good_id")
    zhekoudata.registerTempTable("zhekoudata")
    val frame1 = sqlContext.sql(
      """
        |select
        |*,
        |CAST(priceText as double) priceTexts
        |from
        |zhekoudata
      """.stripMargin).drop("priceText").withColumnRenamed("priceTexts", "priceText")
    frame1.where("priceText > 0 ")
      .registerTempTable("zhekoudata")
    val zhekoudata1 = sqlContext.sql(
      """
        |select good_id,sellCount,priceText,discount_re_1,
        | discount_re_2,discount_re_3,discount_re_4,
        | cast((log(0.2,discount_re_1)+0.1)*100 as bigint) discount_sump_1,
        | cast((log(0.2,discount_re_2)+0.1)*100 as bigint) discount_sump_2,
        | cast((log(0.2,discount_re_3)+0.1)*100 as bigint) discount_sump_3,
        | cast((log(0.2,discount_re_4)+0.1)*100 as bigint) discount_sump_4,
        | cast((log(0.2,discount_re_1)+0.1)*100 as bigint)+
        | cast((log(0.2,discount_re_2)+0.1)*100 as bigint)+
        | cast((log(0.2,discount_re_3)+0.1)*100 as bigint)+
        | cast((log(0.2,discount_re_4)+0.1)*100 as bigint) sum_dis_sum
        | from zhekoudata
      """.stripMargin).cache()
    zhekoudata1.registerTempTable("source_2")
    //    0.1/0.4
    //    10/40
    //0.1 0.1 0.2+0.1 0.3+0.1
    //10 10 30 40
    val zhekoudata2 = sqlContext.sql(
      """
        |select *,
        |case when sum_dis_sum = 0 or sellCount <= 100 then
        |(
        |  case when sellCount <= 100 then least(discount_re_1,discount_re_2,discount_re_3,discount_re_4)*sellCount*priceText
        |  else
        | sellCount*priceText end
        |) else
        |(
        | cast((discount_sump_1/sum_dis_sum)*sellCount as bigint)*discount_re_1*priceText +
        | cast((discount_sump_2/sum_dis_sum)*sellCount as bigint)*discount_re_2*priceText +
        | cast((discount_sump_3/sum_dis_sum)*sellCount as bigint)*discount_re_3*priceText +
        | cast((discount_sump_4/sum_dis_sum)*sellCount as bigint)*discount_re_4*priceText
        |) end salesAmount
        | from source_2
      """.stripMargin).dropDuplicates("good_id")
    zhekoudata2.registerTempTable("zhekoudata2")
    val s = sqlContext.sql(
      """
        |select
        |*,
        |cast(priceText as double)  priceTexts
        |from
        |zhekoudata2
      """.stripMargin).drop("priceText").withColumnRenamed("priceTexts", "priceText").dropDuplicates("good_id")
    s
  }





}
